package com.cy.springcloudalibaba.transaction.producer;

import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.springframework.messaging.Message;

/**
 * @Description: 事务消息监听器
 * @ClassName: TransactionListenerImpl
 * @Author: chenyu
 * @Date: 2025/3/3 16:20
 */
@RocketMQTransactionListener(txProducerGroup = "myTxProducerGroup",
        corePoolSize = 5, maximumPoolSize = 10)
public class TransactionListenerImpl implements RocketMQLocalTransactionListener {

    /**
     * 执行本地事务
     * @param message
     * @param o
     * @return org.apache.rocketmq.spring.core.RocketMQLocalTransactionState
     * @date 2025/3/3 16:23
    */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
        Object num = message.getHeaders().get("test");
        if ("1".equals(num)) {
            System.out.println("executer: " + new String((byte[]) message.getPayload()) + "unknown");
            //不提交，不回滚，等待回调检查
            return RocketMQLocalTransactionState.UNKNOWN;
        }
        else if ("2".equals(num)) {
            System.out.println("executer: " + new String((byte[]) message.getPayload()) + "rollback");
            return RocketMQLocalTransactionState.ROLLBACK;
        }
        System.out.println("executer: " + new String((byte[]) message.getPayload()) + "commit");
        return RocketMQLocalTransactionState.COMMIT;
    }

    /**
     * 回调检查
     * @param message
     * @return org.apache.rocketmq.spring.core.RocketMQLocalTransactionState
     * @date 2025/3/3 16:25
    */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        System.out.println("check: " + new String((byte[]) message.getPayload()));
        return RocketMQLocalTransactionState.COMMIT;
    }
}
